package org.ring.common.transport;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;

/**
 * @since 2023-11-02 14:16
 */
public class NettyClientConnection implements Connection {

    private static final Logger logger = LoggerFactory.getLogger(NettyClientConnection.class);

    private final Channel channel;

    private final String connectionId;

    /**
     * attachment object
     */
    private volatile Object attachment;

    protected NettyClientConnection(Channel channel) {
        this.channel = channel;
        this.connectionId = channel.remoteAddress().toString() + "->" + channel.localAddress().toString();
    }

    @Override
    public CompletableFuture<Object> send(Message request) {
        CompletableFuture future = RpcCompletableFuture.newFuture(request);
        ChannelFuture channelFuture = channel.writeAndFlush(request);
        channelFuture.addListener(f -> {
            if (f.isSuccess()) {
                logger.debug("Send ok------> {}", f.isDone());
            } else {
                future.completeExceptionally(f.cause());
            }
        });
        return future;
    }

    @Override
    public CompletableFuture<Object> request(Object request, int timeout, ExecutorService executor) {
        Message message = new Message();
        message.setData(request);
        return send(message);
    }

    @Override
    public boolean isAlive() {
        return this.channel.isActive();
    }

    @Override
    public String connectionId() {
        return this.connectionId;
    }

    @Override
    public String toString() {
        return "AbstractNettyConnection{" + "channel=" + channel + '}';
    }

    @Override
    public Object attachment() {
        return attachment;
    }

    @Override
    public void attachment(Object attachmentObj) {
        this.attachment = attachmentObj;
    }
}
